package com.mate.cloud.rocketmq.producer;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.mate.cloud.domian.OrderMessage;
import com.mate.cloud.domian.OrderMessageDTO;
import com.mate.cloud.mapper.OrderMessageMapper;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.apache.rocketmq.spring.core.RocketMQTemplate;
import org.apache.rocketmq.spring.support.RocketMQHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.messaging.support.MessageBuilder;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.time.LocalDateTime;

/**
 * 生产者配置（含事务消息、延迟消息）
 * @author:         MI
 * @email:          448341911@qq.com
 * @createTime:     2025/8/12 16:04
 * @updateUser:     MI
 * @updateTime:     2025/8/12 16:04
 * @updateRemark:   修改内容
 * @version:        1.0
 */
@Slf4j
@Service
public class OrderProducer {
    @Autowired
    private RocketMQTemplate rocketMQTemplate;

    @Autowired
    private OrderMessageMapper orderMessageMapper;

    // 发送普通顺序消息（按orderId哈希选择队列）
    public void sendOrderMessage(OrderMessageDTO message) {
        rocketMQTemplate.syncSendOrderly(
                "order-topic",
                MessageBuilder.withPayload(message).build(),
                message.getOrderId()  // Sharding Key
        );
        log.info("发送顺序消息: {}", message);
    }

    // 发送延迟消息（如订单超时未支付）
    public void sendDelayMessage(OrderMessageDTO message, int delayLevel) {
        message.setDelayLevel(delayLevel);
        rocketMQTemplate.syncSend(
                "order-topic",
                MessageBuilder.withPayload(message)
                        .setHeader(RocketMQHeaders.DELAY, delayLevel)
                        .build()
        );
        log.info("发送延迟消息，延迟级别{}: {}", delayLevel, message);
    }

    // 发送事务消息（创建订单）
    @Transactional
    public void sendTransactionMessage(OrderMessageDTO message) {
        // 1. 执行本地事务（保存订单）
        OrderMessage order = new OrderMessage();
        order.setOrderId(message.getOrderId());
        order.setStatus("CREATED");
        order.setCreateTime(LocalDateTime.now());
        orderMessageMapper.insert(order);

        // 2. 发送事务消息
        rocketMQTemplate.sendMessageInTransaction(
                "order-topic",
                MessageBuilder.withPayload(message).build(),
                order
        );
        log.info("发送事务消息: {}", message);
    }

    // 事务监听器（回调检查本地事务状态）
    @RocketMQTransactionListener
    public class TransactionListenerImpl implements RocketMQLocalTransactionListener {
        @Override
        public RocketMQLocalTransactionState executeLocalTransaction(Message msg, Object arg) {
            try {
                OrderMessage order = (OrderMessage) arg;
                // 模拟业务检查
                if (order != null) {
                    return RocketMQLocalTransactionState.COMMIT;
                }
                return RocketMQLocalTransactionState.ROLLBACK;
            } catch (Exception e) {
                return RocketMQLocalTransactionState.ROLLBACK;
            }
        }

        @Override
        public RocketMQLocalTransactionState checkLocalTransaction(Message msg) {
            // 检查本地事务是否成功（如查询数据库）
            OrderMessageDTO dto = (OrderMessageDTO) msg.getPayload();
            // 构建查询条件：判断是否存在 orderId 等于 dto.getOrderId() 的记录
            QueryWrapper<OrderMessage> queryWrapper = new QueryWrapper<>();
            queryWrapper.eq("order_id", dto.getOrderId()); // 假设数据库字段名为 order_id
            // 调用 exists 方法，传入查询条件
            boolean exists = orderMessageMapper.exists(queryWrapper);
            return exists ? RocketMQLocalTransactionState.COMMIT : RocketMQLocalTransactionState.ROLLBACK;
        }
    }

}
